package com.guagua.util;

import static me.prettyprint.hector.api.factory.HFactory.createColumn;
import static me.prettyprint.hector.api.factory.HFactory.createColumnQuery;
import static me.prettyprint.hector.api.factory.HFactory.createMultigetSliceQuery;
import static me.prettyprint.hector.api.factory.HFactory.createMutator;
import static me.prettyprint.hector.api.factory.HFactory.getOrCreateCluster;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import me.prettyprint.cassandra.model.ConfigurableConsistencyLevel;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.service.FailoverPolicy;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.HConsistencyLevel;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.Serializer;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.beans.Rows;
import me.prettyprint.hector.api.exceptions.HectorException;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import me.prettyprint.hector.api.query.ColumnQuery;
import me.prettyprint.hector.api.query.MultigetSliceQuery;
import me.prettyprint.hector.api.query.QueryResult;

public class CassandraClient_back {
	
	
	
	private final static String KEYSPACE = getProperties().getProperty("rankinglist_keyspaceName"); // 相当于Database

	private final static String HOST_PORT = getProperties().getProperty("rankinglist_conn_url");

	private final static String CF_NAME = getProperties().getProperty("rankinglist_columnFamily"); // 相当于Table
	
//	private 

	/** Column name where values are stored */
	private final static String COLUMN_NAME = "v";

	private final StringSerializer serializer = StringSerializer.get();

	private final Keyspace keyspace;

	
	private static Properties _systemProperties = null;
	
	private static Properties getProperties(){
		if(_systemProperties == null){
			_systemProperties= new Properties();
			try {
				_systemProperties.load(HazelcastConfigUtil.class.getResourceAsStream("/system.properties"));
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return _systemProperties;
	}
	/**
	 *
	 * @author xylon/2012-8-7/下午12:10:44
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		Cluster c = getOrCreateCluster("Guagua Data Cluster", HOST_PORT); // cluster相当于集群
		
//		Keyspace ks = HFactory.createKeyspace(KEYSPACE, c); // 连接到一个keyspace
		
		ConfigurableConsistencyLevel ccl = new ConfigurableConsistencyLevel();
		ccl.setDefaultWriteConsistencyLevel(HConsistencyLevel.ONE);
		Map accessMap = new HashMap();
		accessMap.put("username", "vas_room_writer");
		accessMap.put("password", "mJhtRf7uLkI");		
		Keyspace ks = HFactory.createKeyspace(KEYSPACE, c, ccl, FailoverPolicy.ON_FAIL_TRY_ALL_AVAILABLE, accessMap); // 连接到一个keyspace

//		ColumnFamilyDefinition cfd = HFactory.createColumnFamilyDefinition(KEYSPACE, CF_NAME, ComparatorType.BYTESTYPE);
//		KeyspaceDefinition myKs = HFactory.createKeyspaceDefinition(KEYSPACE, ThriftKsDef.DEF_STRATEGY_CLASS, 1, Arrays.asList(cfd));
//		c.addKeyspace(myKs, true);

		CassandraClient_back ed = new CassandraClient_back(ks);
//		ed.insert("test_201208071430", "testvalue1", StringSerializer.get());


//		Map<String,String> kv = new HashMap<String,String>();
//		for(int i=0; i < 101; i++){
//			kv.put("test20120807"+Integer.toString(i), Integer.toString(i*3));
//		}
//		ed.insertMulti(kv, StringSerializer.get());
		
//		ed.delete(StringSerializer.get(), "user_goods_info_chg_id");
//		ed.delete(StringSerializer.get(), "user_goods_info_init");
//		ed.delete(StringSerializer.get(), "init_chg_id");
//		ed.insert("init_chg_id", "0", StringSerializer.get());
//		ed.insert("user_goods_info_init", "true", StringSerializer.get());
		
//		ed.delete(StringSerializer.get(), "recv_uids_30_370001");
		
//		ed.delete(StringSerializer.get(), "user_goods_info_chg_id");
//		ed.delete(StringSerializer.get(), "user_goods_info_init");
//		ed.delete(StringSerializer.get(), "init_chg_id");
		
//		for(int i=0; i < 101; i++){
//		System.out.println("test20120807"+Integer.toString(i)+"  :"+ed.get("test20120807"+Integer.toString(i), StringSerializer.get()));
//		}
		
		
		System.out.println("----------------------收入-------------------------------");
		System.out.println(ed.get("recv_uids_1_397718", StringSerializer.get()));
//		System.out.println(ed.get("recv_uids_7_397718", StringSerializer.get()));
//		System.out.println(ed.get("recv_uids_30_397718", StringSerializer.get()));
//		System.out.println("----------------------消费-------------------------------");
//		System.out.println(ed.get("consume_uids_1_397718", StringSerializer.get()));
//		System.out.println(ed.get("consume_uids_7_397718", StringSerializer.get()));
//		System.out.println(ed.get("consume_uids_30_397718", StringSerializer.get()));
		
	
		System.out.println("----------------------用户物品信息-------------------------------");
		
		System.out.println(ed.get("560517", StringSerializer.get()));
//		System.out.println(ed.get("20346", StringSerializer.get()));
//		System.out.println(ed.get("35001313", StringSerializer.get()));
//		System.out.println(ed.get("35448877", StringSerializer.get()));
//		System.out.println(ed.get("900000008", StringSerializer.get()));
		
		System.out.println(ed.get("35449142", StringSerializer.get()));
//		ed.insert("user_goods_info_chg_id", "0", StringSerializer.get());
		System.out.println(ed.get("user_goods_info_init", StringSerializer.get()));
		System.out.println(ed.get("init_chg_id", StringSerializer.get()));
		System.out.println(ed.get("user_goods_info_chg_id", StringSerializer.get()));
		
		
	}

	
	public CassandraClient_back(Keyspace keyspace) {
		this.keyspace = keyspace;
	}
	
	/**
	 * Insert a new value keyed by key
	 * 
	 * @param key
	 *            Key for the value
	 * @param value
	 *            the String value to insert
	 */
	public <K> void insert(final K key, final String value, Serializer<K> keySerializer) {
		createMutator(keyspace, keySerializer).insert(key, CF_NAME, createColumn(COLUMN_NAME, value, serializer, serializer));
	}

	/**
	 * Get a string value.
	 * 
	 * @return The string value; null if no value exists for the given key.
	 */
	public <K> String get(final K key, Serializer<K> keySerializer) throws HectorException {
		ColumnQuery<K, String, String> q = createColumnQuery(keyspace, keySerializer, serializer, serializer);
		QueryResult<HColumn<String, String>> r = q.setKey(key).setName(COLUMN_NAME).setColumnFamily(CF_NAME).execute();
		HColumn<String, String> c = r.get();
		return c == null ? null : c.getValue();
	}

	/**
	 * Get multiple values
	 * 
	 * @param keys
	 * @return
	 */
	public <K> Map<K, String> getMulti(Serializer<K> keySerializer, K... keys) {
		MultigetSliceQuery<K, String, String> q = createMultigetSliceQuery(keyspace, keySerializer, serializer, serializer);
		q.setColumnFamily(CF_NAME);
		q.setKeys(keys);
		q.setColumnNames(COLUMN_NAME);

		QueryResult<Rows<K, String, String>> r = q.execute();
		Rows<K, String, String> rows = r.get();
		Map<K, String> ret = new HashMap<K, String>(keys.length);
		for (K k : keys) {
			HColumn<String, String> c = rows.getByKey(k).getColumnSlice().getColumnByName(COLUMN_NAME);
			if (c != null && c.getValue() != null) {
				ret.put(k, c.getValue());
			}
		}
		return ret;
	}

	/**
	 * Insert multiple values
	 */
	public <K> void insertMulti(Map<K, String> keyValues, Serializer<K> keySerializer) {
		Mutator<K> m = createMutator(keyspace, keySerializer);
		for (Map.Entry<K, String> keyValue : keyValues.entrySet()) {
			m.addInsertion(keyValue.getKey(), CF_NAME, createColumn(COLUMN_NAME, keyValue.getValue(), keyspace.createClock(), serializer, serializer));
		}
		m.execute();
	}

	/**
	 * Delete multiple values
	 */
	public <K> void delete(Serializer<K> keySerializer, K... keys) {
		Mutator<K> m = createMutator(keyspace, keySerializer);
		for (K key : keys) {
			m.addDeletion(key, CF_NAME, COLUMN_NAME, serializer);
		}
		m.execute();
	}
}